package com.fudan.biz.etl.container;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.io.PrintStream;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import com.fudan.biz.etl.entity.EtlDetail;
import com.fudan.biz.etl.entity.EtlTask;
import com.fudan.biz.etl.service.EtlService;
import com.fudan.biz.etl.service.EtlTaskService;
import com.fudan.cfg.RunConfiguration;
import com.fudan.framework.utils.SpringUtils;
import com.fudan.run.JobRunner;

/**
 * 本地任务的容器
 *
 * @author lobster
 *
 */
public class LocalTaskContainer {
	private volatile static LocalTaskContainer instance = new LocalTaskContainer();

	public static void clear() {
			try {
				if(instance.output != null) {
					instance.getOutput().close();
				}
				if(instance.workThread!=null) {
					instance.workThread.shutdown();
				}
			} catch (Exception e) {
				//e.printStackTrace();
			}
		instance.setStatus(null);
	}

	public static LocalTaskContainer apply(EtlDetail detail) throws ContainerOccupiedException {
		if (instance.getStatus() == null) {
			instance.detail = detail;
			instance.startTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
			instance.id = UUID.randomUUID().toString().replaceAll("-", "");
			instance.status = "ready";
			instance.config = instance.detail.getConfig();
			instance.output = new ByteArrayOutputStream();
			instance.workThread = Executors.newSingleThreadExecutor();
			return instance;
		}
		throw instance.new ContainerOccupiedException();
	}

	public static LocalTaskContainer apply(String config) throws ContainerOccupiedException {
		if (instance.getStatus() == null) {
			instance.startTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
			instance.id = UUID.randomUUID().toString().replaceAll("-", "");
			instance.status = "ready";
			instance.config = config;
			instance.output = new ByteArrayOutputStream();
			instance.workThread = Executors.newSingleThreadExecutor();
			return instance;
		}
		throw instance.new ContainerOccupiedException();
	}

	public static LocalTaskContainer get(String id) {
		if(instance.id != null && instance.id.equals(id)) {
			return instance;
		}
		return null;
	}

	public void start() {
		if("ready".equals(instance.status)) {
			EtlService etlService = SpringUtils.getBean(EtlService.class);
			EtlTaskService etlTaskService = SpringUtils.getBean(EtlTaskService.class);
			EtlTask etlTask = new EtlTask();
			if(instance.detail != null) {
				etlTask.setTaskId(this.id);
				instance.detail.setState(1);
				instance.detail.setApplicationId(this.id);
				etlService.update(instance.detail);
				etlTask.setOccurTime(new Date());
				etlTask.setTaskConfig(getConfig());
				etlTask.setTaskMode("local");
				etlTask.setTaskName(this.detail.getName());
			}
			workThread.execute(()->{
				PrintStream out = new PrintStream(getOutput(),true);
				JobRunner jobRunner = null;
				try {
					RunConfiguration configuration = RunConfiguration.buildConfigurationFromJson(instance.getConfig());
					configuration.setRunId(this.id);
					jobRunner = new JobRunner().setCtx(out)
					.buildRunEnv(configuration)
					.buildDataset(configuration)
					.doTrans(configuration)
					.doAction(configuration);
					if(instance.detail != null) {
						instance.detail.setState(2);
						etlService.update(instance.detail);
					}
					out.println("执行成功");
					System.out.println("finish!");
					etlTask.setResult(getOutput().toString());
				} catch (Exception e) {
					if(instance.detail != null) {
						instance.detail.setState(-1);
						etlService.update(instance.detail);
					}
					e.printStackTrace(out);
					e.printStackTrace();
					etlTask.setError(getOutput().toString());
					etlTask.setResult("执行出错");
				}finally {
					instance.status = null;
					jobRunner.closeCtx();
				}
				if(etlTask.getTaskId() != null) {
					etlTaskService.add(etlTask);
				}
			});
		}
	}

	private ExecutorService workThread;
	private String id;
	private String startTime;
	private String status;
	private String config;
	private EtlDetail detail;
	private OutputStream output;

	public String getId() {
		return id;
	}

	public void setId(String id) {
		this.id = id;
	}

	public String getStartTime() {
		return startTime;
	}

	public void setStartTime(String startTime) {
		this.startTime = startTime;
	}

	public String getStatus() {
		return status;
	}

	public void setStatus(String status) {
		this.status = status;
	}

	public String getConfig() {
		return config;
	}

	public void setConfig(String config) {
		this.config = config;
	}

	public OutputStream getOutput() {
		return output;
	}

	public void setOutput(OutputStream output) {
		this.output = output;
	}

	class ContainerOccupiedException extends Exception {
		private static final long serialVersionUID = 1L;
	}
}
